package com.study.flink.java.day05_state;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * flink容错性之重启策略
 */
public class RestartStrategies {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 设定重启策略，默认策略无限重启，中间states数据保存到taskmanager内存中
        env.enableCheckpointing(5000);

        // 设定重启策略(重启间隔时间)，确保程序的健壮性
        // 方式：默认无限重启、根据重启次数、处理失败比例重启
        env.getConfig().setRestartStrategy(org.apache.flink.api.common.restartstrategy.RestartStrategies.fixedDelayRestart(3, 2000));

        DataStream<String> lines = env.socketTextStream("node02", 8888);

        SingleOutputStreamOperator<Tuple2<String, Integer>> wordAndOne = lines.map(new MapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> map(String s) throws Exception {
                if (s == null || "".equals(s) || s.startsWith("error")) {
                    throw new RuntimeException("error!!!shutdown!!!");
                }
                return Tuple2.of(s.toUpperCase(), 1);
            }
        });
        SingleOutputStreamOperator<Tuple2<String, Integer>> summed = wordAndOne.keyBy(0).sum(1);
        summed.print();
        env.execute("RestartStrategiesDemo");
    }
}
